package docsim;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashSet;
import java.util.Iterator;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.util.StringUtils;

public class Reduce1 extends MapReduceBase implements org.apache.hadoop.mapred.Reducer<Text, LongWritable, Text, DoubleWritable> {

	private static LongWritable n;
	public static LongWritable D=new LongWritable(0);


	@Override
	public void reduce(Text key, Iterator<LongWritable> values,
			OutputCollector<Text, DoubleWritable> output, Reporter reporter)
			throws IOException {
		// TODO Auto-generated method stub
		
		HashSet<Long> unique = new HashSet<Long>();
		while(values.hasNext()) {
			unique.add(values.next().get());
		}
		n = new LongWritable(unique.size());
		//D = new LongWritable(1000); // get total number of documents
		
		double idf = Math.log(((double)D.get())/ n.get());
		output.collect(key, new DoubleWritable(idf));	
	}
	
	 @Override
     public void configure(JobConf job) {
         
         if (job.getBoolean("docsim.docids.map", true)) {
             Path[] docidMapFile = null;
             try {
               docidMapFile = DistributedCache.getLocalCacheFiles(job);
             } catch (IOException ioe) {
               System.err.println("Caught exception while getting cached files: " + StringUtils.stringifyException(ioe));
             }
             for (Path docidFile : docidMapFile) {
               //if(docidFile.getName().matches("docid_map.txt"))
               {
            	   try {
					BufferedReader fis = new BufferedReader(new FileReader(docidFile.toString()));
					long d=0;
					while(fis.readLine()!=null)
					{
						d++;
					}
					D = new LongWritable(d);
				} catch (FileNotFoundException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
               }
              	
             }
           }
       }
}